package com.edata.bigdata.streaming.kafka

import com.edata.bigdata.annotations.Edata_Producer
import com.edata.bigdata.streaming.Producer
import org.apache.kafka.clients.producer.ProducerConfig
import org.apache.kafka.common.serialization.StringSerializer
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

import java.util.Properties

@Edata_Producer(target = "StreamingKafkaProducer")
class SKProducer[K,V] extends Producer[V,ProducerSinks[K,V]] with SKConnector {
  private val kafkaParam: Properties=new Properties()
  override var sourceType: String = "KAFKA"
  override var session: SparkSession = _
  override var producer: Broadcast[ProducerSinks[K, V]] = _

  override def create(): Unit = {
    if (producer == null) {
      kafkaParam.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KF_PRO_BOOTSTRAP)
      kafkaParam.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getName)
      kafkaParam.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getName)
      val sc = session.sparkContext
      synchronized {
        if (producer == null) {
          producer = sc.broadcast(ProducerSinks[K, V](kafkaParam))
        }
      }
    }

  }

  override def sendData(rdd: RDD[V],args:String*): Unit = {
    val prod = producer.value
    rdd.foreach(record => {
      prod.send(args(0), record)
    })
  }
}